package com.apache.flink.basic

import com.apache.flink.bean.Domain.Access
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.types.Row

// 开发时要做的第一件事情
import org.apache.flink.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api._


object DataStreamTableSQLApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnvironment = StreamTableEnvironment.create(env)

    val stream = env.readTextFile("data/access.log")
      .map(x => {
        val splits = x.split(",")
        Access(splits(0).trim.toLong, splits(1).trim, splits(2).trim.toDouble)
      })//.print()

    // datastream ==> table
    val accessTable: Table = tableEnvironment.fromDataStream(stream)
    tableEnvironment.createTemporaryView("access", accessTable)

    val resultTable: Table = tableEnvironment.sqlQuery("select * from access where domain='ruozedata.com'")


//    // table ==> datastream
    tableEnvironment.toAppendStream[Row](resultTable).print("==1==")
//    tableEnvironment.toAppendStream[(Long,String,Double)](resultTable).print("==2==")
//    resultTable.toAppendStream[(Long,String,Double)].print("==3==")


//    accessTable.select("*").toAppendStream[Row].print("==所有字段==")
//    accessTable.select("domain").toAppendStream[Row].print("==单个字段过时写法==")
//    accessTable.select('domain).toAppendStream[Row].print("==单个字段推荐写法==")
//
//    accessTable.select("time,domain,traffic").toAppendStream[Row].print("==多个字段过时写法==")
//    accessTable.select('time,'domain,'traffic).toAppendStream[Row].print("==多个字段推荐写法==")


    val accessTable1: Table = tableEnvironment.fromDataStream(stream, 'a, 'b, 'c)
    accessTable1.printSchema()
    accessTable1.select('a, 'b, 'c).toAppendStream[Row].print()

    env.execute(getClass.getCanonicalName)
  }
}
